package cn.itcast.zt;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.json.JSONUtil;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author zhangtian105
 * @ClassName CompleteFutureDemo.java
 * @projectName Java8
 * @Description CompleteFutureDemo
 * @createTime 2020年05月15日 17:17:00
 */
public class CompleteFutureDemo {
    static class CustomThreadFactory implements ThreadFactory {
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(Thread.currentThread().getThreadGroup(), r,"CUSTOMER_DEFAULT_PREFIX_" + threadNumber.getAndIncrement(),0);
            return thread ;
        }
    }

    public static void main(String[] args) throws Exception {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10,
                20,
                5,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(20),
                new CustomThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy()) ;
        // 线程启动时立即激活核心线程
        threadPoolExecutor.prestartAllCoreThreads() ;
        // 是否允许核心线程超时回收
        threadPoolExecutor.allowCoreThreadTimeOut(true);

        System.out.println("************************ 开始执行多线程处理 **************************");
        List<NullBean> nullBeans = new ArrayList<>() ;
        NullBean nullBean1 = NullBean.builder().name("zhangsan").age(12).build() ;
        NullBean nullBean2 = NullBean.builder().name("zhangsan").age(13).build() ;
        NullBean nullBean3 = NullBean.builder().name("zhangsan").age(14).build() ;
        NullBean nullBean4 = NullBean.builder().name("zhangsan").age(15).build() ;
        NullBean nullBean5 = NullBean.builder().name("zhangsan").age(16).build() ;
        NullBean nullBean6 = NullBean.builder().name("zhangsan").age(17).build() ;
        NullBean nullBean7 = NullBean.builder().name("zhangsan").age(18).build() ;

        nullBeans.add(nullBean1) ;
        nullBeans.add(nullBean2) ;
        nullBeans.add(nullBean3) ;
        nullBeans.add(nullBean4) ;
        nullBeans.add(nullBean5) ;
        nullBeans.add(nullBean6) ;
        nullBeans.add(nullBean7) ;

        List<NullBean> beans1 = new ArrayList<>() ;
        List<NullBean> beans2 = new ArrayList<>() ;
        for (NullBean bean : nullBeans) {
            beans1.add(bean) ;
            beans2.add(bean) ;
        }

        CompletableFuture cf1 = CompletableFuture.runAsync(()->{
            Iterator<NullBean> iterator = beans1.iterator();
            while (iterator.hasNext()) {
                NullBean aa = iterator.next() ;
                if (aa.getAge() < 14 || aa.getAge() > 15) {
                    iterator.remove();
                }
            }

            System.out.println("cf1 使用线程>>>>>:" + Thread.currentThread().getName());
        }, threadPoolExecutor) ;

        CompletableFuture cf2 = CompletableFuture.runAsync(()->{
            Iterator<NullBean> iterator = beans2.iterator();
            while (iterator.hasNext()) {
                NullBean bb = iterator.next() ;
                if (bb.getAge() > 16) {
                    iterator.remove();
                }
            }

            System.out.println("cf2 使用线程>>>>>:" + Thread.currentThread().getName());
        }, threadPoolExecutor) ;

        // 这里使用的是不同的线程
        CompletableFuture.allOf(cf1, cf2).join() ;

        nullBeans.clear();

        // 这里有个思路：使用不同的容器会取不同的数据（同一数据来源结构），之后汇聚结果（去掉了使用并发容器以及锁）
        for (NullBean bean : CollUtil.intersection(beans1, beans2)) {
            nullBeans.add(bean) ;
        }

        System.out.println("NullBean数据打印>>>>>:" + nullBeans);

        System.out.println("====================== 测试线程异步 ====================");
        // 使用一个子线程
        CompletableFuture aCf = CompletableFuture.runAsync(()->{
            System.out.println("我独立于主线程执行......" + Thread.currentThread().getName());
        }, threadPoolExecutor) ;

        CompletableFuture bCf = CompletableFuture.supplyAsync(()->{
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "我是bCf异步可返回数据......" + Thread.currentThread().getName() ;
        }, threadPoolExecutor) ;

        CompletableFuture cCf = CompletableFuture.supplyAsync(()->{
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "我是cCf异步可返回数据......" + Thread.currentThread().getName() ;
        }, threadPoolExecutor) ;

        // 使用不同的线程
        bCf.thenCombineAsync(cCf, (a, b) -> {
            System.out.println("thenCombineAsync:AND关系:" + a.toString() + "=>" + b.toString() + ">>>>>:当前线程"+Thread.currentThread().getName());
            return null ;
        }, threadPoolExecutor) ;

        // 使用不同的线程
        bCf.thenAcceptBothAsync(cCf, (a, b) -> {
            System.out.println("thenAcceptBothAsync:AND关系:" + a.toString() + "=>" + b.toString() + ">>>>>:当前线程"+Thread.currentThread().getName());
        }, threadPoolExecutor) ;

        bCf.runAfterBothAsync(cCf, () -> {
            System.out.println("runAfterBothAsync:OR关系:" + ">>>>>:当前线程"+Thread.currentThread().getName());
        }, threadPoolExecutor) ;

        bCf.acceptEitherAsync(cCf, (a) -> {
            System.out.println("acceptEitherAsync:" + a.toString() + ">>>>>:当前线程"+Thread.currentThread().getName());
        }, threadPoolExecutor) ;

        bCf.thenAcceptAsync((a) -> {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.println("thenAcceptAsync:串行执行:" + a.toString() + ">>>>>:当前线程"+Thread.currentThread().getName());
        }, threadPoolExecutor) ;

        Thread.sleep(10000); // 主线程不能结束 或者汇聚

        System.out.println("*********************");
        NullBean nullBeanA = NullBean.builder().name("zhangsan").age(12).build() ;
        NullBean nullBeanB = NullBean.builder().name("lisi").age(32).build() ;
        NullBean nullBeanC = NullBean.builder().name("wangwu").age(18).build() ;
        NullBean nullBeanD = NullBean.builder().name("zhaoliu").age(16).build() ;
        NullBean nullBeanE = NullBean.builder().name("qianqi").age(32).build() ;
        NullBean nullBeanF = NullBean.builder().name("songba").age(26).build() ;
        NullBean nullBeanG = NullBean.builder().name("fujiu").age(23).build() ;

        List<NullBean> beanListRes = ListUtil.list(false) ;
        List<NullBean> beanList = ListUtil.list(false) ;
        beanList.add(nullBeanA) ;
        beanList.add(nullBeanB) ;
        beanList.add(nullBeanC) ;
        beanList.add(nullBeanD) ;
        beanList.add(nullBeanE) ;
        beanList.add(nullBeanF) ;
        beanList.add(nullBeanG) ;

        CompletableFuture.runAsync(() -> {
            beanListRes.addAll(beanList.parallelStream().sorted(Comparator.comparing(NullBean::getAge).reversed()).collect(Collectors.toList())) ;
        }, threadPoolExecutor).join() ;

        System.out.println(JSONUtil.toJsonStr(beanListRes));

        long startTime = System.currentTimeMillis() ;
        CompletableFuture<Boolean> rcsCF = CompletableFuture.supplyAsync(() ->  {
            try {
                Thread.sleep(5000);
            }catch (Exception ex) {

            }
            return true ;
        }) ;

        CompletableFuture<Boolean> jrCF =  CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(15000);
            }catch (Exception ex) {

            }

            return false ;
        }) ;

        try {
            System.out.println(rcsCF.get());
        }catch (Exception ex) {

        }

        try {
            System.out.println(jrCF.get());
        }catch (Exception ex) {

        }

        System.out.println(System.currentTimeMillis() - startTime);
    }
}
